package com.syf.kit.soa.core.send;

import com.syf.core.service.context.ServiceRequest;
import com.syf.core.service.context.ServiceResponse;
import com.syf.kit.SyfJsonMapper;
import com.syf.kit.soa.SOARegionDocument;
import com.syf.kit.soa.config.SySOAClientConfiguration;
import com.syf.kit.soa.core.SOAExchangeData;
import com.syf.kit.soa.core.SOARegionInfo;
import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.SyfSOAProtocolCont;
import com.syf.kit.soa.core.balance.SyfRpcBalance;
import com.syf.kit.soa.core.send.protocol.TcpClient;
import com.syf.kit.soa.core.send.protocol.client.PooledTcpClient;
import com.syf.kit.soa.core.send.protocol.client.UnPooledTcpClient;
import org.springframework.core.io.support.SpringFactoriesLoader;

import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

public class SyfRpcSender {

    /***
     * 按service region instance（ip:port）的tcpClient map
     * */
    private final Map<String, TcpClient> regionInstanceClientMap = new ConcurrentHashMap<>();

    /**
     * service region 自动发现
     */
    private SOARegionDocument regionDocument;

    /**
     * 负载均衡器
     */
    private SyfRpcBalance balance;

    private SySOAClientConfiguration configuration;

    private final SyfJsonMapper jsonMapper;

    public SyfRpcSender(SOARegionDocument regionDocument, SySOAClientConfiguration configuration) {
        this.regionDocument = regionDocument;
        this.configuration = configuration;
        this.jsonMapper = new SyfJsonMapper(true, true);

        // 通过spi获取balance
        List<SyfRpcBalance> syfRpcBalances = SpringFactoriesLoader.loadFactories(SyfRpcBalance.class, SyfRpcSender.class.getClassLoader());
        if (syfRpcBalances == null || syfRpcBalances.isEmpty()){
            throw new RuntimeException("SyfRpcBalance的SPI配置有误，请核实！");
        }

        balance = syfRpcBalances.get(0);
    }

    /**
     * 调用远程的service
     *
     * @param regionName    目标service region名称
     * @param serviceCode   交易码
     * @param request       SERVICE REQUEST
     * @param respDataClass 响应数据体的类
     */
    public <R> ServiceResponse<R> callRemoteService(String regionName, String serviceCode, ServiceRequest request, Class<R> respDataClass) {
        TcpClient tcpClient = getTcpClient(regionName);

        // 将ServiceRequest打包成SyfRpcMsg
        SyfRpcMsg respMsg = tcpClient.send(packMsg(serviceCode, request, SyfSOAProtocolCont.REQUEST_RESPONSE_REQ_TYPE), configuration.getReqTimeoutMs());
        return jsonMapper.toBean(respMsg.getBizData(), ServiceResponse.class, respDataClass);
    }

    /**
     * 调用远程的service
     *
     * @param regionName  目标service region名称
     * @param serviceCode 交易码
     * @param request     SERVICE REQUEST
     */
    public String callRemoteService(String regionName, String serviceCode, ServiceRequest request) {
        TcpClient tcpClient = getTcpClient(regionName);

        // 将ServiceRequest打包成SyfRpcMsg
        SyfRpcMsg respMsg = tcpClient.send(packMsg(serviceCode, request, SyfSOAProtocolCont.REQUEST_RESPONSE_REQ_TYPE), configuration.getReqTimeoutMs());
        return respMsg.getBizData();
    }

    /**
     * 触发远程的service
     *
     * @param regionName  目标service region名称
     * @param serviceCode 交易码
     * @param request     SERVICE REQUEST
     */
    public void triggerRemoteService(String regionName, String serviceCode, ServiceRequest request) {
        TcpClient tcpClient = getTcpClient(regionName);
        tcpClient.send(packMsg(serviceCode, request, SyfSOAProtocolCont.TRIGGER_REQ_TYPE), configuration.getReqTimeoutMs());
    }

    /**
     * 将ServiceRequest打包为SyfRpcMsg
     */
    private SyfRpcMsg packMsg(String serviceCode, ServiceRequest request, byte reqType) {
        SyfRpcMsg reqMsg = new SyfRpcMsg();

        reqMsg.setMagicNum(configuration.getMagicNum());
        reqMsg.setVersion(SyfSOAProtocolCont.RPC_VERSION);
        reqMsg.setReqType(reqType);
        reqMsg.setBizDataType(SyfSOAProtocolCont.JSON_BIZ_DATA_TYPE);
        reqMsg.setCallCount(request.getHead().getCallCount() + 1);
        UUID uuid = UUID.randomUUID();
        reqMsg.setReqUUID(uuid.getLeastSignificantBits());
        reqMsg.setServiceCode(serviceCode);
        reqMsg.setReserved(new byte[4]);

        SOAExchangeData exchangeData = new SOAExchangeData();
        exchangeData.setTraceId(request.getHead().getTraceId());
        exchangeData.setBizOrderNo(request.getHead().getBizOrderNo());
        exchangeData.setBizUseId(request.getHead().getBizUseId());
        exchangeData.setBizReqJson(jsonMapper.toJsonString(request.getDataObj()));
        reqMsg.setBizData(jsonMapper.toJsonString(exchangeData));

        return reqMsg;
    }

    /**
     * 获取Rpc的tcpClient
     */
    private TcpClient getTcpClient(String regionName) {
        // 服务发现
        List<SOARegionInfo> soaRegionInfos = regionDocument.discoveryService(regionName);

        if (soaRegionInfos == null || soaRegionInfos.isEmpty()) {
            throw new RuntimeException("未找到对应service region:[" + regionName + "]的注册信息");
        }

        // 负载
        SOARegionInfo targetRegionInfo = balance.balance(soaRegionInfos);

        String clientKey = getRegionClientKey(targetRegionInfo.getHost(), targetRegionInfo.getListenPort());
        // 本地如果没有此service region instance对应的tcp client，需要新建
        if (regionInstanceClientMap.get(clientKey) == null) {

            TcpClient client = buildTcpClient(targetRegionInfo.getHost(), targetRegionInfo.getListenPort());
            regionInstanceClientMap.put(clientKey, client);
        }

        return regionInstanceClientMap.get(clientKey);
    }

    /**
     * rpc instance key
     */
    private String getRegionClientKey(String host, int port) {
        return host.concat(":").concat(String.valueOf(port));
    }

    /**
     * 新建tcpClient
     */
    public TcpClient buildTcpClient(String host, int port) {
        TcpClient client = null;

        if (configuration.getUsePooledConn()) {
            client = new PooledTcpClient(host, port, configuration.getMagicNum(), configuration.getConnTimeoutMs(), configuration.getConnPoolMaxSize(), configuration.getConnPoolIdleSize(), configuration.getBeatInterval());
        } else {
            client = new UnPooledTcpClient(host, port, configuration.getConnTimeoutMs());
        }

        return client;
    }

    /**
     * 关闭
     */
    public void close() {
        for (TcpClient tcpClient : regionInstanceClientMap.values()) {
            tcpClient.close();
        }
    }

    public SyfJsonMapper getJsonMapper() {
        return jsonMapper;
    }
}
